package com.vpen.client.test;

import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.converter.StringMessageConverter;
import org.springframework.messaging.simp.stomp.DefaultStompSession;
import org.springframework.messaging.simp.stomp.StompHeaders;
import org.springframework.messaging.simp.stomp.StompSession;
import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.socket.WebSocketHttpHeaders;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.messaging.WebSocketStompClient;

import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 描述:<a href="https://stackoverflow.com/questions/41604828/how-to-use-and-customize-messageconversionspring-websocket-client">接收不到订阅消息问题</a>
 *
 * @author 韦鹏
 * @date 2022/11/26 20:56
 */
@Slf4j
public class StompClientApp {

    public static final String URL = "ws://localhost:8000/chat-endpoint";


    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 创建客户端
        StandardWebSocketClient client = new StandardWebSocketClient();
        // 创建Stomp客户端
        WebSocketStompClient socketStompClient = new WebSocketStompClient(client);

        /**
         * 使用多种消息转换器
         * */
        List<MessageConverter> converters = new ArrayList<>();
        converters.add(new MappingJackson2MessageConverter());
        converters.add(new StringMessageConverter());

        // 设置消息转换器：否则消息无法接收到
        socketStompClient.setMessageConverter(new CompositeMessageConverter(converters));

        // 设置定时线程池
        socketStompClient.setTaskScheduler(taskScheduler());

        // 握手用到的headers
        WebSocketHttpHeaders handshakeHeaders = new WebSocketHttpHeaders();
        handshakeHeaders.add("ipAddr", "192.168.0.1");

        // stomp连接用到的headers，可以放token
        StompHeaders connectHeaders = new StompHeaders();
        connectHeaders.add("name", "vpen");


        // 连接websocket服务器
        ListenableFuture<StompSession> future = socketStompClient.connect(URL, handshakeHeaders, connectHeaders, new MyStompSessionHandler());

        // 回调方法
        future.addCallback(new MyListenableFutureCallback());
        StompSession stompSession = future.get();

        stompSession.setAutoReceipt(true);
        stompSession.subscribe("/topic/hello", new MyStompSessionHandler());


        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()) {
            String str = scanner.nextLine();
            stompSession.send("/app/hello1", str);
        }
    }

    public static TaskScheduler taskScheduler() {
        // ThreadPoolTaskScheduler是TaskScheduler接口的实现，包装 原生的ScheduledThreadPoolExecutor。
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setPoolSize(3);
        // 设置默认线程名称
        taskScheduler.setThreadNamePrefix("stomp-task-");
        // 设置拒绝策略1
        taskScheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 等待所有任务结束后再关闭线程池
        taskScheduler.setWaitForTasksToCompleteOnShutdown(true);
        taskScheduler.initialize();
        return taskScheduler;
    }
}
